package com.atguigu.rabbitmq.four;

import com.atguigu.rabbitmq.utils.RabbitMqutils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.util.Date;
import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * 验证发布确认模式
 * <p>
 * 1.单个确认
 * <p>
 * 使用时间 比较那种确认方式最好 发布1000个消息单独确认消息，耗时5976ms
 * <p>
 * <p>
 * 2.批量确认
 * 发布1000个消息批量确认消息，耗时362ms
 * 3.异步确认
 * 发布1000个消息异步发布确认消息，耗时73ms
 * 发布1000个消息异步发布确认消息，耗时59ms
 */
public class ConfirmMessage {

    //批量发消息的个数
    public static final int MESSAGE_COUNT = 1000;


    public static void main(String[] args) throws Exception {
        //  1.单个确认
        //  ConfirmMessage.publishMessageIndividuallu();
        //    2.批量确认
        //   ConfirmMessage.publishMessageBatch();
        //  3.异步确认
        ConfirmMessage.publishMEssageAsync();
    }


    //单个确认
    public static void publishMessageIndividuallu() throws Exception {
        Channel channel = RabbitMqutils.getChannel();
        //队列的声明
        String quenuName = UUID.randomUUID().toString(); //javaJDK提供的一个自动生成主键的方法
        System.out.println("quenuName-+++" + quenuName);
        channel.queueDeclare(quenuName, true, false, false, null);

        //开启确认发布的，模式
        channel.confirmSelect();
        //记录时间 开始的时间
        long l = System.currentTimeMillis();//毫秒的


        //批量的发消息
        //MESSAGE_COUNT.fori 就可以出现下面了省略的写法
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", quenuName, null, message.getBytes());
            //单个消息确认
            boolean flag = channel.waitForConfirms();
            if (flag) {
                System.out.println("消息发送成功");
            }
        }
        //结束时间 开始的时间
        long e = System.currentTimeMillis();//毫秒的
        System.out.println("发布" + MESSAGE_COUNT + "个消息单独确认消息，耗时" + (e - l) + "ms");
    }


    //批量发布确认
    public static void publishMessageBatch() throws Exception {

        Channel channel = RabbitMqutils.getChannel();
        //队列的声明
        String quenuName = UUID.randomUUID().toString();
        System.out.println("quenuName-+++" + quenuName);
        channel.queueDeclare(quenuName, true, false, false, null);

        //开启确认发布的，模式
        channel.confirmSelect();
        //记录时间 开始的时间
        long l = System.currentTimeMillis();//毫秒的


        //批量确认的长度
        int batshSize = 100;


        //未确认的个数
//批量发送消息 批量发布确认
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = i + "";
            channel.basicPublish("", quenuName, null, message.getBytes());
            //判断达到消息100条消息的时候 批量确认一次
            if (i % batshSize == 0) {
                //消息确认
                channel.waitForConfirms();
            }
        }

        //结束时间 开始的时间
        long e = System.currentTimeMillis();//毫秒的
        System.out.println("发布" + MESSAGE_COUNT + "个消息批量确认消息，耗时" + (e - l) + "ms");


    }


    //异步发布确认
    public static void publishMEssageAsync() throws Exception {
        Channel channel = RabbitMqutils.getChannel();
        //队列的声明
        String quenuName = UUID.randomUUID().toString(); //javaJDK提供的一个自动生成主键的方法
        System.out.println("quenuName-+++" + quenuName);
        channel.queueDeclare(quenuName, true, false, false, null);

        //开启确认发布的模式
        channel.confirmSelect();
        /**
         * 线程安全由于的一个 哈希表 适用于到并发的情况下
         * 选择他的理由
         * 1.他可以轻松的 将序号和消息关联 long ,String
         * 2.可以轻松的批量的删除条目  只要给我key就可以删除value  ()
         *3.支持高并发 （多线程）
         */
        ConcurrentSkipListMap<Long,String> outstandingConfirms=
                new ConcurrentSkipListMap<>();

        //消息确认成功 回调 函数
        ConfirmCallback ackCallback = (var1, var3) -> {
            //2.删除掉确认的消息 剩下就是未确认的消息
    if(var3){
        //是批量的话 调用id删除
        ConcurrentNavigableMap<Long, String> confirmed =
                outstandingConfirms.headMap(var1);
        confirmed.clear();
    }else{
  //不是的话 直接删除
        outstandingConfirms.remove(var1);

    }
            System.out.println("确认的消息" + var1);

        };
        //消息确认失败 回调 函数
        /**
         * 1.消息的标记  var1
         * 、2.是否批量删除  var3  true/false
         */
        ConfirmCallback nackCallback = (var1, var3) -> {
            //3.打印未确认的消息有哪些
            String messgae = outstandingConfirms.get(var1);
            System.out.println("未确认的消息是"+messgae+"----未确认的消息tag" + var1);
        };

        //准备消息的监听器 监听纳西人消息成功了 那些消息失败了呢
        /**
         * 1.监听成功的 监听那些是成的消息
         * 2.监听失败的消息，监听那些消息是失败的
         */
        channel.addConfirmListener(ackCallback, nackCallback); //异步监听

        //记录时间 开始的时间
        long l = System.currentTimeMillis();//毫秒的
        //批量发送消息
        for (int i = 0; i < MESSAGE_COUNT; i++) {
            String message = "消息" + i;
            channel.basicPublish("", quenuName, null, message.getBytes());
            //1.此处要记录下所有要发送消息的 总和
            // 序号  信息
            outstandingConfirms.put(channel.getNextPublishSeqNo(),message);

        }


        //结束时间 开始的时间
        long e = System.currentTimeMillis();//毫秒的
        System.out.println("发布" + MESSAGE_COUNT + "个消息异步发布确认消息，耗时" + (e - l) + "ms");

    }


}
